package com.elvin.platform;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.elvin.platform.beans.AlarmRule;
import com.elvin.platform.beans.Device;
import com.elvin.platform.beans.HistoryData;
import com.elvin.platform.rabitMQ.consumeService.ReceiveVo;
import com.elvin.platform.service.AlarmRuleService;
import com.elvin.platform.service.DeviceService;
import com.elvin.platform.service.HistoryDataService;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Log4j2
//@Configuration
public class MqttConfig {

  @Autowired
  private DeviceService deviceService;
  @Autowired
  private HistoryDataService historyDataService;
  /**
   * 1、先创建连接
   */

  /**
   * 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置。
   * @return factory
   */
  @Bean
  public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();

    // 设置代理端的URL地址，可以是多个
    options.setServerURIs(new String[]{"tcp://127.0.0.1:1883","tcp://81.68.121.222:1883","tcp://81.68.121.222:8083","ws://81.68.121.222:8083/mqtt"});

    factory.setConnectionOptions(options);
    return factory;
  }


  /**
   * 2、入站通道
   */
  @Bean
  public MessageChannel mqttInputChannel() {
    return new DirectChannel();
  }


  /**
   * 入站
   */
  @Bean
  public MessageProducer inbound() {
    // Paho客户端消息驱动通道适配器，主要用来订阅主题
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("consumerClient-paho",
        mqttClientFactory(), "boat", "testtopic_r", "t/a", "+/sensor");
    adapter.setCompletionTimeout(5000);

    // Paho消息转换器
    DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
    // 按字节接收消息
//        defaultPahoMessageConverter.setPayloadAsBytes(true);
    adapter.setConverter(defaultPahoMessageConverter);
    adapter.setQos(1); // 设置QoS
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
  }


  /**
   * 3、消息转化，中间站
   */
  @Bean
  // ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
  @ServiceActivator(inputChannel = "mqttInputChannel")
  public MessageHandler handler() {
    return message -> {

      String payload = message.getPayload().toString();

      // byte[] bytes = (byte[]) message.getPayload(); // 收到的消息是字节格式
      String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
      // 根据主题分别进行消息处理。
      if (topic.matches(".+/sensor")) { // 匹配：1/sensor
        String sensorSn = topic.split("/")[0];
        System.out.println("传感器" + sensorSn + ": 的消息： " + payload);
      } else if (topic.equals("collector")) {
        System.out.println("采集器的消息：" + payload);
      }else if (topic.equals("t/a")) {
        JSONObject map = JSONObject.parseObject(payload);
        log.info("t/a收到的消息为："+map.toString());
        //收到消息之后进行其他数据处理
        //1.根据 dev_plateform=='LoRa' 和  dev_id查找设备
        // 修改设备表中的设备，将data放到report1中
        EntityWrapper<Device> wrapper = new EntityWrapper<>();
        wrapper.eq("dev_platform", "LoRa");
        wrapper.eq("dev_id", map.get("devid"));
        if(!StringUtils.isEmpty(map.get("devid"))){
          List<Device> deviceList = deviceService.selectList(wrapper);
          if (deviceList.size()>0) {
            Device device1 = new Device();
            device1.setReport1((String)map.get("data"));
            for (Device device : deviceList) {
              device1.setId(device.getId());
              deviceService.updateById(device1);
            }
          }
          //2.新增到 report_data表，新dev_code,report,repoet_date,type
          HistoryData historyData = new HistoryData();
          historyData.setDevId((String)map.get("devid"));
          historyData.setReport((String)map.get("data"));
          historyData.setDataSources("lora_pressure");
          historyData.setCreateTime(new Date());
          historyDataService.insert(historyData);
        }
      } else {
        System.out.println("丢弃消息：主题[" + topic  + "]，负载：" + payload);
      }

    };
  }


  /**
   * 4、消息出去
   */
  /**
   * 出站通道
   */
  @Bean
  public MessageChannel mqttOutboundChannel() {
    return new DirectChannel();
  }


  /**
   * 出站
   */
  @Bean
  @ServiceActivator(inputChannel = "mqttOutboundChannel")
  public MessageHandler outbound() {

    // 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("publishClient", mqttClientFactory());
    messageHandler.setAsync(true); // 如果设置成true，即异步，发送消息时将不会阻塞。
    messageHandler.setDefaultTopic("command");
    messageHandler.setDefaultQos(1); // 设置默认QoS
    // Paho消息转换器
    DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();

    // defaultPahoMessageConverter.setPayloadAsBytes(true); // 发送默认按字节类型发送消息
    messageHandler.setConverter(defaultPahoMessageConverter);
    return messageHandler;
  }



}
